package com.huawei.spare.part.replace.rx;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.disposables.Disposable;
import io.reactivex.schedulers.Schedulers;

import java.util.concurrent.TimeUnit;

public class RxDebounce {

    public static void main(String[] args) {
        final Disposable subscribe = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.setCancellable(() -> System.out.println("canceled"));

                int i = 0;
                while (i++ < 40) {
                    emitter.onNext(i);
                    try {
                        Thread.sleep(100);
                    } catch (Exception e) {
                        System.out.println("interupt");
                    }
                }

                emitter.onComplete();
            }
        }).throttleFirst(1, TimeUnit.SECONDS).doOnDispose(() -> {
            System.out.println("dispose 2");
        }).doOnComplete(() -> {
            System.out.println("complete");
        }).subscribeOn(Schedulers.io()).observeOn(Schedulers.computation()).subscribe((n) -> {
            System.out.println(n);
        });


        try {
            Thread.sleep(1000);
            System.out.println(subscribe.isDisposed());
            Thread.sleep(200000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
